package com.gitee.xhs.savePointReader.keyedState.mapStateReader;

import com.gitee.xhs.keyedState.mapState.MapStateConstant;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.state.api.functions.KeyedStateReaderFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.Map;

/**
 * @author BigHead
 * @date 2022/6/30 16:42
 * @description MapStateReaderFunction
 */
public class MapStateReaderFunction extends KeyedStateReaderFunction<Long, MapStateExhibitor> {

    private final String jobId;
    private final String operatorUid;
    private transient MapState<Integer, Double> mapState;

    public MapStateReaderFunction(String jobId, String operatorUid) {
        this.jobId = jobId;
        this.operatorUid = operatorUid;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        mapState = getRuntimeContext().getMapState(MapStateConstant.MAP_STATE_DESCRIPTOR);
    }

    @Override
    public void readKey(Long key, Context ctx, Collector<MapStateExhibitor> out) throws Exception {
        MapStateExhibitor mapStateExhibitor = new MapStateExhibitor();
        mapStateExhibitor.setJobId(jobId);
        mapStateExhibitor.setOperatorUid(operatorUid);
        mapStateExhibitor.setKey(key);
        Map<Integer, Double> tempMap = new HashMap<>();
        for (Map.Entry<Integer, Double> mapEntry : mapState.entries()) {
            tempMap.put(
                    mapEntry.getKey(),
                    mapEntry.getValue()
            );
        }
        mapStateExhibitor.setMapStateValue(tempMap);

        out.collect(mapStateExhibitor);
    }
}
